package io.mft.commons.db;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.hutool.core.bean.BeanUtil;
import io.mft.commons.function.response.kv.KVMap;
import io.mft.commons.multithreading.worker.Work;
import io.mft.commons.multithreading.worker.WorkManager;
/**
 * 异步批量新增
 * @说明：TODO
 * @类型名称：AsyncBatchDbExecutor
 * @创建者: 胡常建  
 * @创建时间: 2021年9月15日 上午10:56:22
 * @修改者: 胡常建  
 * @修改时间: 2021年9月15日 上午10:56:22
 */
public class AsyncBatchDbExecutor <T>{
	/**
	 *   数据行Db保存监听
	 * @说明：TODO
	 * @类型名称：DbRowInstanceSavedListener
	 * @创建者: 胡常建  
	 * @创建时间: 2019年10月14日 下午5:02:30
	 * @修改者: 胡常建  
	 * @修改时间: 2019年10月14日 下午5:02:30
	 */
	public static interface DbRowInstanceSaveListener{
		
		void onSaved(int result,KVMap row);
		void onSaving(KVMap row);
	}
	
	private static final Logger logger = LoggerFactory.getLogger(AsyncBatchDbExecutor.class);
	private int batchSize=1000;
	private String tbName;
	private String dataSourceId;
	private boolean genId=false;
	private DbRowInstanceSaveListener dbRowInstanceSaveListener;
	public void setDbRowInstanceSaveListener(DbRowInstanceSaveListener dbRowInstanceSaveListener) {
		this.dbRowInstanceSaveListener = dbRowInstanceSaveListener;
	}
	private static  HashMap<String, AsyncBatchDbExecutor<?>> map=new HashMap<String, AsyncBatchDbExecutor<?>>();
	@SuppressWarnings("unchecked")
	public static <RT> AsyncBatchDbExecutor<RT> getInstance(Class<?>  owner,String aliasId,String dataSourceId,String tbName,int batchSize,int batchWorkerNumber,boolean genId) {
		String id=AsyncBatchDbExecutor.class.getName()+"-"+aliasId+"-"+owner.getName()+"-"+dataSourceId+"-"+tbName;
		if (map.containsKey(id)) {
			AsyncBatchDbExecutor<?> asyncBatchDbExecutor= map.get(id);
			return  (AsyncBatchDbExecutor<RT>)asyncBatchDbExecutor;
		}
		synchronized (map) {
			if (!map.containsKey(id)) {
				AsyncBatchDbExecutor<RT> instance=new AsyncBatchDbExecutor<RT>(owner,aliasId,dataSourceId,tbName,batchSize,batchWorkerNumber,genId);
				map.put(id, instance);
			}
		}
		AsyncBatchDbExecutor<?> asyncBatchDbExecutor= map.get(id);
		return  (AsyncBatchDbExecutor<RT>)asyncBatchDbExecutor;
	}
	private AsyncBatchDbExecutor(Class<?>  owner,String aliasId,String dataSourceId,String tbName,int batchSize,int batchWorkerNumber,boolean genId) {
		this.tbName=tbName;
		this.batchSize=batchSize;
		this.dataSourceId=dataSourceId;
		this.genId=genId;
		String key=AsyncBatchDbExecutor.class.getName()+"-"+aliasId+"-"+owner.getName()+"-"+dataSourceId+"-"+tbName;
		WorkManager workManager=WorkManager.getManager(key, batchWorkerNumber, batchWorkerNumber);
		for (int i = 0; i < batchWorkerNumber; i++) {
			workManager.acceptWork(work);
		}
	}
	private BlockingQueue<T> queue=new LinkedBlockingQueue<T>();
	public int number() {
		return queue.size();
	}
	public void add(T object) {
		try {
			queue.add(object);
		} catch (Exception e) {
			logger.error("",e);
		}
	}
	
	private Work work=new Work() {
		
		@Override
		public Object handle() {
			List<KVMap> parameters=new ArrayList<KVMap>();
			while (true) {
				try {
					T object=queue.poll();
					if (object!=null) {
						KVMap data=new KVMap();
						if (object instanceof Map) {
							Map map=(Map)object;
							data.putAll(map);
							data=DbUtil.formatToUnderline(data);
						}else {
							Map<String, Object> kv=BeanUtil.beanToMap(object, true, true);
							data.putAll(kv);
						}
						if (dbRowInstanceSaveListener!=null) {
							dbRowInstanceSaveListener.onSaving(data);
						}
						parameters.add(data);
					}
					if (parameters.size()==batchSize||object==null) {
						if (parameters.size()>0) {
							try {
								int[] results=BatchExecuteUtil.execute(dataSourceId, tbName, parameters);
								if (results==null) {
									continue;
								}
								int i=0;
								for (int s : results) {
									try {
										KVMap kv=parameters.get(i);
										if (dbRowInstanceSaveListener!=null) {
											dbRowInstanceSaveListener.onSaved(s,kv);
										}
									} catch (Exception e) {
										logger.error("",e);
									}
									i++;
								}
							} catch (Exception e) {
								//queue.addAll(objects);
								logger.error("",e);
							}finally {
								parameters.clear();
							}
						}
						parameters.clear();
					}
					if (object==null) {
						TimeUnit.SECONDS.sleep(3);
					}
				} catch (Exception e) {
					logger.error("数据源["+dataSourceId+"]-数据表["+tbName+"]异步批量操作异常",e);
				}
			}
		}
	};
}
